package com.ym.spring.rabbit;

import cn.hutool.core.exceptions.ExceptionUtil;
import com.alibaba.fastjson.JSONObject;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.GetResponse;
import com.ym.core.exception.BusinessException;
import com.ym.core.rabbit.AbstractDelayedTask;
import com.ym.core.rabbit.IMessageSenderSV;
import com.ym.core.rabbit.MessageTypeEnum;
import com.ym.core.rabbit.NormalMessageTask;
import com.ym.medical.service.IYunxinUserSV;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.connection.Connection;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.ObjectInputStream;
import java.nio.charset.Charset;
import java.util.Map;

/**
 * 消费MQ消息
 *
 * @author: 李涛
 * @version: 2019年09月18日 10:41
 */
@Component
@EnableRabbit
public class RabbitMessageHandle {

	private static final Logger LOG = LoggerFactory.getLogger(RabbitMessageHandle.class);

	@Autowired
	private IMessageSenderSV messageSenderSV;

	@Autowired
	private IYunxinUserSV yunxinUserSV;

	@Autowired
	private ConnectionFactory connectionFactory;

	@Value("${config-center.rabbitRuleConfig.deadExchange}.que.nml.${config-center.rabbitRuleConfig.systemMark}")
	private String normalDeadQueue;

	@Value("${config-center.rabbitRuleConfig.deadExchange}.que.dly.${config-center.rabbitRuleConfig.systemMark}")
	private String delayDeadQueue;

	/**
	 * 普通消息
	 */
	@RabbitListener(queues = "${config-center.rabbitRuleConfig.normalExchange}.que.${config-center.rabbitRuleConfig.systemMark}", containerFactory = "rabbitListenerContainerFactory")
	public void autoNormalMsg(@Payload byte[] body, @Headers Map<String, Object> headers, Channel channel) throws Exception {
		channel.txSelect();
		try {
			NormalMessageTask messageTask = JSONObject.parseObject(new String(body, Charset.defaultCharset()), NormalMessageTask.class);
			boolean success = normalHandle(messageTask);
			if (success) {
				channel.basicAck((long) headers.get(AmqpHeaders.DELIVERY_TAG), false);
			} else {
				channel.basicReject((long) headers.get(AmqpHeaders.DELIVERY_TAG), false);
			}
		} finally {
			channel.txCommit();
		}
	}

	/**
	 * 手动消费普通消息
	 */
	public boolean manualConsumptionNormal() throws Exception {
		boolean success = false;
		Channel channel = null;
		Envelope envelope = null;
		try {
			Connection connection = connectionFactory.createConnection();
			channel = connection.createChannel(true);
			channel.txSelect();
			GetResponse response = channel.basicGet(normalDeadQueue, false);
			envelope = response.getEnvelope();
			byte[] body = response.getBody();
			NormalMessageTask normalMessageTask = JSONObject.parseObject(new String(body, Charset.defaultCharset()), NormalMessageTask.class);
			success = normalHandle(normalMessageTask);
			if (success) {
				channel.basicAck(envelope.getDeliveryTag(), false);
			} else {
				channel.basicReject(envelope.getDeliveryTag(), true);
			}
		} finally {
			channel.txCommit();
		}
		return success;
	}

	/**
	 * 消费普通消息方法
	 *
	 * @param messageTask
	 */
	private boolean normalHandle(NormalMessageTask messageTask) {
		try {
			MessageTypeEnum messageTypeEnum = messageTask.getMessageTypeEnum();
			Object msg = messageTask.getMsg();
			LOG.info("消费消息 [ {} ],消息ID为[ {} ]", messageTypeEnum.getDescribe(), messageTask.getUuid());
			switch (messageTypeEnum) {
				case YUN_XIN: {
					yunxinUserSV.syncMessages((String) msg);
				}
				break;
				default: {
					// do
					LOG.info("未知消息:{}", (String) msg);
				}
			}
		} catch (Exception e) {
			if (e instanceof BusinessException) {
				LOG.info(e.getMessage());
				return true;
			} else {
				LOG.error("消费异常:{}", ExceptionUtil.getMessage(e));
				return false;
			}
		}
		return true;
	}


	/**
	 * 延时消息,执行策略
	 * <p>
	 * 能收到说明已经到时间了
	 */
	@RabbitListener(queues = "${config-center.rabbitRuleConfig.delayExchange}.que.${config-center.rabbitRuleConfig.systemMark}", containerFactory = "rabbitListenerContainerFactory")
	public void autoDelayMsg(@Payload Message message, @Headers Map<String, Object> headers, Channel channel) throws Exception {
		channel.txSelect();
		try {
			boolean success = delayHandle(message);
			if (success) {
				channel.basicAck((long) headers.get(AmqpHeaders.DELIVERY_TAG), false);
			} else {
				channel.basicReject((long) headers.get(AmqpHeaders.DELIVERY_TAG), false);
			}
		} finally {
			channel.txCommit();
		}
	}

	/**
	 * 消费延时消息方法
	 *
	 * @param message
	 */
	private boolean delayHandle(Message message) {
		MessageProperties messageProperties = message.getMessageProperties();
		Map<String, Object> headers = messageProperties.getHeaders();
		try {
			// 判定是否要多次延时
			long nextDelay = (long) headers.get("next_delay");
			if (nextDelay > 0) {
				messageSenderSV.sendAgain(message, nextDelay);
				return true;
			}
			byte[] body = message.getBody();
			if (body != null && body.length > 0) {
				//判定为一个有效消息，进行执行
				try (ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(body));) {
					AbstractDelayedTask abstractDelayedTask = (AbstractDelayedTask) ois.readObject();
					LOG.info("执行延时任务 [ {} ],消息ID为[ {} ],参数为:{}", abstractDelayedTask.getDescribe(), abstractDelayedTask.getUuid(), JSONObject.toJSONString(headers));
					abstractDelayedTask.excute();
				}
			}
			return true;
		} catch (Throwable e) {
			if (e instanceof BusinessException) {
				LOG.info(e.getMessage());
				return true;
			} else {
				LOG.error("消费异常:{}", ExceptionUtil.getMessage(e));
				return false;
			}
		}
	}

	/**
	 * 手动消费延时消息
	 */
	public boolean manualConsumptionDelay() throws Exception {
		Channel channel = null;
		Envelope envelope = null;
		boolean success = true;
		try {
			Connection connection = connectionFactory.createConnection();
			channel = connection.createChannel(true);
			channel.txSelect();
			GetResponse response = channel.basicGet(delayDeadQueue, false);
			envelope = response.getEnvelope();
			byte[] body = response.getBody();
			AbstractDelayedTask abstractDelayedTask = toTask(body);
			abstractDelayedTask.excute();
		} catch (Exception e) {
			if (e instanceof BusinessException) {
				LOG.info(e.getMessage());
				channel.basicAck(envelope.getDeliveryTag(), false);
			} else {
				channel.basicReject(envelope.getDeliveryTag(), true);
				LOG.error("消费异常:{}", ExceptionUtil.getMessage(e));
				success = false;
			}
		} finally {
			channel.txCommit();
		}
		return success;
	}

	/**
	 * 数组转对象
	 *
	 * @param bytes
	 * @return
	 */
	private AbstractDelayedTask toTask(byte[] bytes) {
		AbstractDelayedTask obj = null;
		try {
			ByteArrayInputStream bis = new ByteArrayInputStream(bytes);
			ObjectInputStream ois = new ObjectInputStream(bis);
			obj = (AbstractDelayedTask) ois.readObject();
			ois.close();
			bis.close();
		} catch (IOException ex) {
			ex.printStackTrace();
		} catch (ClassNotFoundException ex) {
			ex.printStackTrace();
		}
		return obj;
	}
}
